import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, NamespaceDescriptor, TableName}
import org.apache.logging.log4j.scala.Logging

import scala.jdk.CollectionConverters._
import scala.util.{Failure, Success, Using}

object HBasePractice extends Logging {

  // namespace、table name、cf 描述信息
  private val ns = "zkx"
  private val cf1 = "info"
  private val cf2 = "score"
  private val table: TableName = TableName.valueOf(ns, "student")

  def main(args: Array[String]): Unit = {
    val conf = HBaseConfiguration.create(new Configuration())
    // 使用 Using 提供类似于 try-with-resource 的功能
    Using(ConnectionFactory.createConnection(conf)) { con =>
      // 建表
      createTable(con)
      // 插入数据
      insertData(con)
      // 查询数据
      scanData(con)
    } match {
      case Success(_) => logger.info("程序运行结束...")
      case Failure(exception) => logger.error(exception)
    }
  }

  private final def createTable(con: Connection): Unit = {
    Using.resource(con.getAdmin) { admin =>
      // namespace 不存在则创建
      if (!admin.listNamespaceDescriptors().exists(_.getName.equals(ns))) {
        logger.info(s"创建 namespace: $ns")
        admin.createNamespace(NamespaceDescriptor.create(ns).build())
      }
      // table 不存在则创建
      if (!admin.tableExists(table)) {
        logger.info(s"创建 table: ${table.toString}")
        admin.createTable(TableDescriptorBuilder
          .newBuilder(table)
          .setColumnFamilies(Seq(cf1, cf2).map(cf => ColumnFamilyDescriptorBuilder.newBuilder(cf.getBytes()).setMaxVersions(3).build).asJava)
          .build)
      }
    }
  }

  private final def insertData(con: Connection): Unit = {
    logger.info("插入数据...")
    // 使用 mutator 批量插入
    Using.resource(con.getBufferedMutator(table)) { mutator =>
      mutator.mutate(
        Seq(
          // info:Student_id
          StudentPut("Tom", cf1, "student_id", "20210000000001").asPut,
          StudentPut("Jerry", cf1, "student_id", "20210000000002").asPut,
          StudentPut("Jack", cf1, "student_id", "20210000000003").asPut,
          StudentPut("Rose", cf1, "student_id", "20210000000004").asPut,
          StudentPut("zkx", cf1, "student_id", "G20210735010429").asPut,
          // info:class
          StudentPut("Tom", cf1, "class", "1").asPut,
          StudentPut("Jerry", cf1, "class", "1").asPut,
          StudentPut("Jack", cf1, "class", "2").asPut,
          StudentPut("Rose", cf1, "class", "2").asPut,
          StudentPut("zkx", cf1, "class", "5").asPut,
          // score:understanding
          StudentPut("Tom", cf2, "understanding", "75").asPut,
          StudentPut("Jerry", cf2, "understanding", "85").asPut,
          StudentPut("Jack", cf2, "understanding", "80").asPut,
          StudentPut("Rose", cf2, "understanding", "60").asPut,
          StudentPut("zkx", cf2, "understanding", "100").asPut,
          // score:programming
          StudentPut("Tom", cf2, "programming", "82").asPut,
          StudentPut("Jerry", cf2, "programming", "67").asPut,
          StudentPut("Jack", cf2, "programming", "80").asPut,
          StudentPut("Rose", cf2, "programming", "61").asPut,
          StudentPut("zkx", cf2, "programming", "100").asPut
        ).asJava
      )
      mutator.flush()
    }
  }

  private final def scanData(con: Connection): Unit = {
    logger.info("查询数据...")
    Using.resource(con.getTable(table)) { table =>
      def printTable(table: Table, cf: String): Unit = {
        table.getScanner(cf.getBytes)
          .asScala
          .map(rs => (rs.getRow, rs.getNoVersionMap))
          .foreach { fqv =>
            fqv._2.asScala
              .flatten(x => x._2.asScala)
              .foreach { qv => println(Bytes.toString(fqv._1), Bytes.toString(qv._1), Bytes.toString(qv._2)) }
          }
      }

      logger.info(s"查询 $cf1 列族数据:")
      printTable(table, cf1)
      logger.info(s"查询 $cf2 列族数据:")
      printTable(table, cf2)
    }
  }
}

/*
 * 对需要插入的信息进程封装，并提供快速生成 Put 的方法
 */
private case class StudentPut(row: String, cf: String, qualifier: String, value: String) {
  def asPut: Put = {
    val put = new Put(row.getBytes())
    put.addColumn(cf.getBytes(), qualifier.getBytes(), value.getBytes())
    put
  }
}